#include "src/common/mini_container.h"

#include <string.h>

#include "raptor-lite/utils/log.h"
#include "raptor-lite/utils/time.h"

#ifdef _WIN32
#include "src/windows/connection.h"
#include "src/windows/iocp_thread.h"
#else
#include "src/linux/connection.h"
#include "src/linux/epoll_thread.h"
#endif
#include "src/common/connection_id.h"
#include "src/common/endpoint_impl.h"

namespace raptor {

enum MiniContainerMessageType { kIOEvent, kTimeoutChecking, kHeartbeat, kSendData, kCloseEndpoint };

struct MiniContainerMessageNode {
    MultiProducerSingleConsumerQueue::Node node;
    MiniContainerMessageType type;
    uint32_t index;
    Slice s;
    bool notify;
    struct EventDetail detail;
};

MiniContainer::MiniContainer(uint16_t id)
    : shutdown_(true)
    , id_(id)
    , max_count_(0)
    , magic_(0)
    , option_(nullptr) {}

MiniContainer::~MiniContainer() {}

raptor_error MiniContainer::Init(int init_size, int max_size) {
    if (!shutdown_) return RAPTOR_ERROR_FROM_STATIC_STRING("MiniContainer had already initialized");

    RAPTOR_ASSERT(init_size <= max_size);

    mgr_.resize(init_size);

    for (int i = 0; i < init_size; i++) {
        free_list_.emplace_back(i);
        mgr_[i].first = nullptr;
        mgr_[i].second = timeout_records_.end();
    }

    bool success = false;
    thd_ = Thread("MiniContainer",
                  std::bind(&MiniContainer::MessageQueueThread, this, std::placeholders::_1),
                  nullptr, &success);

    if (!success) {
        log_error("MiniContainer: Failed to create message queue thread");
        return RAPTOR_ERROR_FROM_STATIC_STRING("MiniContainer: Failed to create mq threads");
    }

    max_count_ = max_size;
    shutdown_ = false;
    count_.Store(0);
    client_count_.Store(0);
    last_check_time_.Store(0);
    return RAPTOR_ERROR_NONE;
}

raptor_error MiniContainer::Start() {
    if (shutdown_) {
        return RAPTOR_ERROR_FROM_STATIC_STRING("MiniContainer has not been initialized");
    }

    thd_.Start();
    return RAPTOR_ERROR_NONE;
}

void MiniContainer::Shutdown() {
    if (!shutdown_) {
        log_warn("MiniContainer: prepare to shutdown");
        shutdown_ = true;
        cv_.Signal();
        thd_.Join();

        mtx_.Lock();
        for (auto &obj : mgr_) {
            if (obj.first) {
                obj.first->Shutdown(false, nullptr);
                obj.first.reset();
            }
        }
        mgr_.clear();
        free_list_.clear();
        timeout_records_.clear();
        mtx_.Unlock();
        client_count_.Store(0);

        // clear message queue
        bool empty = true;
        do {
            auto n = mpscq_.PopAndCheckEnd(&empty);
            auto msg = reinterpret_cast<MiniContainerMessageNode *>(n);
            if (msg != nullptr) {
                delete msg;
            }
        } while (!empty);
        log_warn("MiniContainer: shutdown");
    }
}

void MiniContainer::EventHandler(EventDetail *detail) {
    if (shutdown_) return;
    auto msg = new MiniContainerMessageNode;
    msg->type = kIOEvent;
    msg->detail = *detail;
    mpscq_.push(&msg->node);
    count_.FetchAdd(1, MemoryOrder::RELAXED);
    cv_.Signal();
}

void MiniContainer::TimeoutChecking() {
    if (shutdown_) return;
    auto msg = new MiniContainerMessageNode;
    msg->type = kTimeoutChecking;
    mpscq_.push(&msg->node);
    count_.FetchAdd(1, MemoryOrder::RELAXED);
    cv_.Signal();
}

void MiniContainer::HeartBeatHandler(uint32_t index) {
    if (shutdown_) return;
    auto msg = new MiniContainerMessageNode;
    msg->type = kHeartbeat;
    msg->index = index;
    mpscq_.push(&msg->node);
    count_.FetchAdd(1, MemoryOrder::RELAXED);
    cv_.Signal();
}

void MiniContainer::SendToEndpoint(uint32_t index, const Slice &s) {
    if (shutdown_) return;
    auto msg = new MiniContainerMessageNode;
    msg->type = kSendData;
    msg->index = index;
    msg->s = s;
    mpscq_.push(&msg->node);
    count_.FetchAdd(1, MemoryOrder::RELAXED);
    cv_.Signal();
}

void MiniContainer::CloseEndpoint(uint32_t index, bool notify) {
    if (shutdown_) return;
    auto msg = new MiniContainerMessageNode;
    msg->type = kCloseEndpoint;
    msg->index = index;
    msg->notify = notify;
    mpscq_.push(&msg->node);
    count_.FetchAdd(1, MemoryOrder::RELAXED);
    cv_.Signal();
}

void MiniContainer::MessageQueueThread(void *) {
    while (!shutdown_) {
        if (count_.Load() < 1) {
            AutoMutex g(&cmtx_);
            cv_.Wait(&cmtx_, 1000);
            continue;
        }
        auto n = mpscq_.pop();
        auto msg = reinterpret_cast<MiniContainerMessageNode *>(n);
        if (msg != nullptr) {
            count_.FetchSub(1, MemoryOrder::RELAXED);
            this->Dispatch(msg);
            delete msg;
        }
    }
}

void MiniContainer::Dispatch(struct MiniContainerMessageNode *msg) {
    switch (msg->type) {
    case kIOEvent:
        this->OnRecvSendEvent(&msg->detail);
        break;
    case kTimeoutChecking:
        this->OnTimeoutChecking();
        break;
    case kHeartbeat:
        this->OnHeartBeat(msg->index);
        break;
    case kSendData:
        this->OnSendData(msg->index, msg->s);
        break;
    case kCloseEndpoint:
        this->OnCloseEndpoint(msg->index, msg->notify);
        break;
    }
}

void MiniContainer::OnSendData(uint32_t index, const Slice &s) {
    if (index < mgr_.size() && mgr_[index].first) {
        mgr_[index].first->SendMsg(s);
    }
}

void MiniContainer::OnCloseEndpoint(uint32_t index, bool notify) {
    if (index < mgr_.size() && mgr_[index].first) {
        mgr_[index].first->Shutdown(notify, RAPTOR_ERROR_NONE);
        DeleteConnection(index);
    }
}

void MiniContainer::OnHeartBeat(uint32_t index) {
    if (index < mgr_.size() && mgr_[index].first) {
        mgr_[index].first->DoHeartBeat();
    }
}

void MiniContainer::OnTimeoutChecking() {

    auto current = GetCurrentMilliseconds();
    if (current - last_check_time_.Load() < 3000) {
        return;
    }

    AutoMutex g(&mtx_);

    last_check_time_.Store(current);

    raptor_error err = RAPTOR_ERROR_FROM_STATIC_STRING("Connection: kMaxTimeoutForReceiving");

    auto it = timeout_records_.begin();
    while (it != timeout_records_.end()) {
        if (it->first > current) {
            break;
        }

        uint32_t index = it->second;

        ++it;

        log_info("ContainerImpl: check_receiving_interval close connection: id=%d, index=%u",
                 (int)id_, index);

        mgr_[index].first->Shutdown(true, err);
        mgr_[index].first.reset();
        timeout_records_.erase(mgr_[index].second);
        mgr_[index].second = timeout_records_.end();
        free_list_.emplace_back(index);
        client_count_.FetchSub(1, MemoryOrder::RELAXED);
    }
}

void MiniContainer::OnRecvSendEvent(EventDetail *detail) {
    uint32_t connection_id =
        static_cast<uint32_t>(reinterpret_cast<intptr_t>(detail->ptr) & 0xffffffff);

    uint32_t index = internal::GetConnectionIndex(connection_id);

    if (index >= mgr_.size()) {
        log_error("MiniContainer: Found an invalid index, cid = %x event = %d",
                  (uint64_t)detail->ptr, detail->event_type);
        return;
    }

    if (detail->event_type & internal::kErrorEvent) {
        OnErrorEvent(index, detail->error_code);
        return;
    }

    if (detail->event_type & internal::kRecvEvent) {
        OnRecvEvent(index, detail);
    }

    if (detail->event_type & internal::kSendEvent) {
        OnSendEvent(index, detail);
    }
}

// Receiver implement (iocp / epoll event)
void MiniContainer::OnErrorEvent(uint32_t index, int error_code) {
    raptor_error err = MakeRefCounted<Status>(error_code, "MiniContainer:OnErrorEvent");

    auto conn = mgr_[index].first;
    if (conn) {
        conn->Shutdown(true, err);
        DeleteConnection(index);
    }
}

void MiniContainer::OnRecvEvent(uint32_t index, EventDetail *detail) {
    auto conn = mgr_[index].first;
    if (!conn) return;

    raptor_error err = conn->DoRecvEvent(detail);
    if (err == RAPTOR_ERROR_NONE) {
        RefreshTime(index);
        return;
    }

    conn->Shutdown(true, err);
    DeleteConnection(index);
}

void MiniContainer::OnSendEvent(uint32_t index, EventDetail *detail) {
    auto conn = mgr_[index].first;
    if (!conn) return;

    raptor_error err = conn->DoSendEvent(detail);
    if (err == RAPTOR_ERROR_NONE) {
        return;
    }

    conn->Shutdown(true, err);
    DeleteConnection(index);
}

std::shared_ptr<Connection> MiniContainer::InsertEndpoint(Endpoint ep) {

    AutoMutex g(&mtx_);

    if (free_list_.empty() && static_cast<int32_t>(mgr_.size()) >= max_count_) {
        log_error("MiniContainer: Failed to Attach Endpoint, MaxCount = %u CurrentSize = %u",
                  max_count_, mgr_.size());
        return nullptr;
    }

    std::shared_ptr<EndpointImpl> obj = ep.impl_;

    if (free_list_.empty()) {
        int32_t count = static_cast<int32_t>(mgr_.size());
        int32_t expand = ((count * 2) < max_count_) ? (count * 2) : max_count_;

        mgr_.resize(expand);
        for (int32_t i = count; i < expand; i++) {
            mgr_[i].first = nullptr;
            mgr_[i].second = timeout_records_.end();
            free_list_.emplace_back(static_cast<uint32_t>(i));
        }
    }

    uint32_t index = free_list_.front();
    free_list_.pop_front();

    uint32_t cid = internal::BuildConnectionId(magic_, id_, index);

    obj->SetConnectionId(cid);

    auto conn = std::make_shared<Connection>(obj);
    conn->Init(index);

    mgr_[index].first = conn;
    if (option_->check_receiving_interval) {
        int64_t deadline = GetCurrentMilliseconds() + option_->max_receiving_interval;
        mgr_[index].second = timeout_records_.insert({deadline, index});
    }

    client_count_.FetchAdd(1, MemoryOrder::RELAXED);

    return conn;
}

void MiniContainer::DeleteConnection(uint32_t index) {
    if (!mgr_[index].first) return;

    AutoMutex g(&mtx_);

    if (option_->check_receiving_interval) {
        timeout_records_.erase(mgr_[index].second);
        mgr_[index].second = timeout_records_.end();
    }
    mgr_[index].first.reset();
    free_list_.emplace_back(index);
    client_count_.FetchSub(1, MemoryOrder::RELAXED);
}

void MiniContainer::RefreshTime(uint32_t index) {
    if (!option_->check_receiving_interval) return;
    if (!mgr_[index].first) return;

    AutoMutex g(&mtx_);

    int64_t deadline = GetCurrentMilliseconds() + option_->max_receiving_interval;
    timeout_records_.erase(mgr_[index].second);
    mgr_[index].second = timeout_records_.insert({deadline, index});
}

int32_t MiniContainer::Count() { return client_count_.Load(); }

} // namespace raptor
